/*
 * =====================================================================================
 *
 *       Filename:  cl_tcp_server.cxx
 *
 *    Description:  
 *
 *        Version:  1.0
 *        Created:  05/05/2011 03:51:00 PM
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Ken Chen (Ken), chen_kenz@emc.com
 *        Company:  EMC
 *
 * =====================================================================================
 */

#ifndef __GXX_EXPERIMENTAL_CXX0X__
#define __GXX_EXPERIMENTAL_CXX0X__
#endif 

#include <cassert>
#include "cl_tcp_server.hxx"
#include "cl_server_socket.hxx"
#include "cl_common_defs.hxx"
#include "cl_utils.hxx"
#include "cl_event_looper.hxx"
#include "cl_logger.hxx"

namespace neo
{
using namespace std;
using namespace std::placeholders;

CLTcpServer::CLTcpServer(const string& host, int port) 
  : m_acceptLooper(new CLEventLooper()),
  m_acceptor(new CLServerSocket(m_acceptLooper.get(), host, port, CL_SOCK_STREAM)),
  m_nextLooper(0),
  m_started(false),
  m_ioThrNum(-1),
	m_connections(clMaxFdSize()) 
{
  m_acceptor->setNewConnCallback(bind(&CLTcpServer::handleNewConnection, this, _1, _2));
}

CLTcpServer::~CLTcpServer() 
{
  ConnectionTab::iterator itr(m_connections.begin());
  ConnectionTab::iterator itrEnd(m_connections.begin());
	for(; itr != itrEnd; ++itr)
	{
    if((*itr).get() != NULL)
		{
      (*itr)->ioLooper()->runJob(bind(&CLTcpConn::handleDestroy, (*itr))); 
		  (*itr).reset();
		}
	}
  CLLOG_C("NET", "%s", "TCP Server shut down");
}

void CLTcpServer::setThreadNum(int numThreads)
{
  m_ioThrNum = numThreads;
}

void CLTcpServer::start(void)
{
  if(m_started)
  {
    CLLOG_C("NET", "%s", "TCP server already started");
    return;
  }
  m_started = true;
  init();
  m_acceptLooper->start();
  if(!m_acceptor->listening())
  {
    m_acceptLooper->runJob(bind(&CLServerSocket::listen, m_acceptor.get(), 128));
  }
  CLLOG_D("NET", "%s", "TCP server started");
}

// Called in the m_acceptLooper thread...
void CLTcpServer::handleNewConnection(int sockfd, CLSocketAddrUPtr peerAddr) 
{
  CLLOG_D("NET", "TCP server accept new connection from: %s", peerAddr->toString().c_str());
  assert(m_acceptLooper->isInLooperThread());
  int connId = nextFreeSlot();
  CLEventLooper* ioLooper = nextIoLooper();
  CLLOG_D("NET", "Allocated connection slot %d to service it", connId); 
  CLTcpConnSPtr conn(new CLTcpConn(ioLooper, connId, sockfd, move(peerAddr)));
  conn->setConnCallback(m_connCallback);
  conn->setMessageCallback(m_msgCallback);
  conn->setWriteCompleteCallback(m_writeCallback);
  conn->setCloseCallback(bind(&CLTcpServer::recycleConnection, this, _1));
	ioLooper->runJob(bind(&CLTcpConn::handleEstablished, conn));
	m_connections[connId] = conn;
}

// Called in the ioLooper thread...
void CLTcpServer::recycleConnection(const CLTcpConnSPtr& conn)
{
	assert(conn->ioLooper()->isInLooperThread());
  m_acceptLooper->runJob(bind(&CLTcpServer::recycleConnectionInLoop, this, conn));
}

// Called in the m_acceptLooper thread...
void CLTcpServer::recycleConnectionInLoop(const CLTcpConnSPtr& conn)
{
	assert(m_acceptLooper->isInLooperThread());
  if(m_closeCallback)
	{
		m_closeCallback(conn);
	}
  int connId = conn->connId();
  m_connections[connId].reset();
  m_freeSlots.push_front(connId);
  CLEventLooper* ioLooper = conn->ioLooper();
  ioLooper->queueJob(bind(&CLTcpConn::handleDestroy, conn));
}

void CLTcpServer::setConnCallback(const CLConnCallback& cb) 
{
  m_connCallback = cb;
}

void CLTcpServer::setMessageCallback(const CLMessageCallback& cb) 
{
  m_msgCallback = cb;
}

void CLTcpServer::setWriteCompleteCallback(const CLWriteCompleteCallback& cb) 
{
  m_writeCallback = cb;
}

void CLTcpServer::setCloseCallback(const CLCloseCallback& cb) 
{
  m_closeCallback = cb;
}

// If m_ioThrNum is less than zero, decide the thread number for
// doing IO according to the cpu numbers.
void CLTcpServer::initIoLooper(void)
{
  if((m_ioThrNum < 0) && ((m_ioThrNum = clOnlineProcessorNum() - 1) < 0))
  {
    m_ioThrNum = 0;
  }

  for(int i = 0; i < m_ioThrNum; ++i)
  {
    m_ioLoopers.push_back(move(unique_ptr<CLEventLooper>(new CLEventLooper)));
  }

  for(int i = 0; i < m_ioThrNum; ++i)
  {
    m_ioLoopers[i]->start();
  }
}

// Pay attention to the memory footprint,
// If found free slot table is using too much
// memory(shoult not), consider adopting other solution.
// Using free slot is just simplying the implementation.
void CLTcpServer::initFreeSlotTab(void)
{
  size_t n = m_connections.size();
  for(size_t i = 0; i < n; ++i)
	{
    m_freeSlots.push_back(i);
	}
}

void CLTcpServer::init(void)
{
	initIoLooper();
	initFreeSlotTab();
}

int CLTcpServer::nextFreeSlot(void)
{
  assert(!m_freeSlots.empty());
  int slotId = m_freeSlots.front();
  m_freeSlots.pop_front();
	return slotId;
}

CLEventLooper* CLTcpServer::nextIoLooper(void) 
{
  if(m_ioThrNum > 0)
  {
    int nextLooper = (m_nextLooper++)%m_ioThrNum;
    return m_ioLoopers[nextLooper].get();
  }
  else
  {
    return m_acceptLooper.get();
  }
}

} // neo

